package priv.bestbeat.reportcenter.admin.monitor.rocket;

import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
import priv.bestbeat.reportcenter.core.monitor.TaskMonitorProxy;
import priv.bestbeat.reportcenter.core.monitor.TaskObservable;
import priv.bestbeat.reportcenter.core.task.states.TaskStateEnum;

import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 张渠钦
 * @date 2022/5/6 14:21
 */
@Component
@RocketMQMessageListener(consumerGroup = "Gtest1",topic = "taskReply",selectorType = SelectorType.TAG,selectorExpression = "*",
        consumeMode = ConsumeMode.CONCURRENTLY,messageModel = MessageModel.CLUSTERING)
public class RocketTaskMonitor implements RocketMQListener<TaskObservable<?>> {

    @Resource
    private TaskMonitorProxy taskMonitorProxy;

    private ConcurrentHashMap<Long,Integer> tryAgainCount = new ConcurrentHashMap<>();

    @Override
    public void onMessage(TaskObservable<?> message) {
        if (TaskStateEnum.TRY_AGAIN.equals(message.getState())) {
            if (tryAgainCount.containsKey(message.getId())) {
                tryAgainCount.computeIfPresent(message.getId(),(key,value) -> ++value);
            } else {
                tryAgainCount.putIfAbsent(message.getId(),0);
            }


            if (tryAgainCount.get(message.getId()) >= 3) {
                message.setState(TaskStateEnum.FAILED);
            }
        }
        taskMonitorProxy.changeState(message);
    }
}
